package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;

import com.atguigu.edu.realtime.common.bean.UserComeBackBean;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DwsUserComeBackWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsUserComeBackWindow()
                .start(13006,4,"DwsUserComeBackWindow","dwd_user_login");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSONObject.parseObject(s);
                    }
                }
        );
        SingleOutputStreamOperator<JSONObject> wmStream = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(
                (v, ts) -> v.getLong("ts")
        ));

        SingleOutputStreamOperator<UserComeBackBean> beanStream = wmStream.keyBy(v->v.getString("userId")).map(
                new RichMapFunction<JSONObject, UserComeBackBean>() {

                    ValueState<String> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("ValueStateDescriptor", String.class);
                        state = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public UserComeBackBean map(JSONObject jsonObject) throws Exception {
                        //从状态获取数据
                        String lastDate = state.value();
                        String today = jsonObject.getString("loginTime");
                        Long ts = jsonObject.getLong("ts");
                        Long userNum = 0L;
                        //如果有数据且上次登录时间是7天前
                        if (lastDate != null) {
                                //DateFormatUtil.dateTimeToTs(today)-DateFormatUtil.dateTimeToTs(lastDate)>7*24*3600*1000) { 为了有效果
                            //是回流用户
                            userNum = 1L;
                        }

                        state.update(today);
                        UserComeBackBean userComeBackBean = UserComeBackBean.builder().userNum(userNum)
                                .build();
                        return userComeBackBean;
                    }
                }
        );

        SingleOutputStreamOperator<UserComeBackBean> resultStream = beanStream.windowAll(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).reduce(
                new ReduceFunction<UserComeBackBean>() {
                    @Override
                    public UserComeBackBean reduce(UserComeBackBean userComeBackBean, UserComeBackBean t1) throws Exception {
                        userComeBackBean.setUserNum(userComeBackBean.getUserNum() + t1.getUserNum());
                        return userComeBackBean;
                    }
                },
                new AllWindowFunction<UserComeBackBean, UserComeBackBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<UserComeBackBean> iterable, Collector<UserComeBackBean> collector) throws Exception {
                        UserComeBackBean next = iterable.iterator().next();
                        next.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        next.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        next.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(next);
                    }
                }
        );
        resultStream.print();
//        resultStream
//                .map(new BeanToJsonStrMapFunction<>())
//                .sinkTo(FlinkSinkUtil.getDorisSink("dws_user_come_back_window"));

    }
}
